package org.porsche.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author junxian.dou
 * @Package：org.porsche
 * @Project：Default (Template) Project
 * @name：${NAME}
 * @Date：2024/7/5 14:06
 * @Filename：${NAME}
 * @verson 1.0.0
 * @desc
 */
//处理器
class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String,String> {

    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println("processor开始处理");
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println("processor拿到数据" + item);
        item+="hello";
        submit(item);
        subscription.request(1);

    }

    @Override
    public void onError(Throwable throwable) {
    }

    @Override
    public void onComplete() {
        System.out.println("processor处理完成");
    }
}

public class Main {

    public static void main(String[] args) {
        //发布者
        SubmissionPublisher<String> stringSubmissionPublisher = new SubmissionPublisher<>();

        //处理器
        MyProcessor myProcessor = new MyProcessor();

        //订阅者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override //事件触发机制
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + "订阅开始了....");
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + "订阅者，接受数据：" + item);
                if (item.equals("5")) {
                    subscription.cancel();
                }
                subscription.request(1);
                if (item.equals("1")) {
                    throw new RuntimeException("数组错误");
                }

            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + "订阅者，发生错误：" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + "订阅者，完成操作...");
            }
        };

        //发布者-》处理器-》订阅者
        stringSubmissionPublisher.subscribe(myProcessor);
        myProcessor.subscribe(subscriber);

        //发布者绑定订阅者
        //stringSubmissionPublisher.subscribe(subscriber);

        for (int i = 0; i < 10; i++) {
            stringSubmissionPublisher.submit(String.valueOf(i));
        }

        stringSubmissionPublisher.close();

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}